// Copyright (c) 2022 by Duguang.IO Inc. All Rights Reserved.
// Author: Ethan Liu
// Date: 2022-05-21 12:47:52

package runtime

import (
	"bytes"
	"context"
	"fmt"
	"github.com/sirupsen/logrus"
	"io"
	"jianmu-worker-kube/client"
	"jianmu-worker-kube/engine"
	"jianmu-worker-kube/livelog"
	"jianmu-worker-kube/types"
	"path"
	"runtime/debug"
	"time"

	logger "jianmu-worker-kube/logging"
)

type Resources struct {
	HelperResources *engine.Resources
	BuildResources  *engine.Resources
}
type Worker struct {
	// Client 负责与CI Server交互的客户端
	Client client.Client
	// Engine Kubernetes客户端
	Engine *engine.Kubernetes
	// Worker ID
	ID string
	// Worker Name
	Name string
	// Worker Type
	Type string
	// pod 资源限制
	Resources *Resources
	Mask      string
}

// Run 运行Runner
func (w *Worker) Run(ctx context.Context, unit *engine.Unit) error {
	livelog := livelog.New(w.Client, "build", unit.TaskInstanceId)
	writer := NewReplacer(livelog, unit.Secrets, w.Mask)
	unit.Log = writer
	defer writer.Close()

	log := logger.FromContext(ctx).
		WithField("unit.name", unit.PodSpec.Name)

	err := w.Client.Accept(ctx, unit.Runners[0])

	if err != nil && err == client.ErrOptimisticLock {
		log.Debug("runner accepted by another worker")
		return nil
	}
	if err != nil {
		log.WithError(err).Error("cannot accept runner")
		return err
	}

	// 如果是VOLUME类型,直接成功
	// TODO 获取node信息回传给server,server后期回传这个信息
	if unit.Type == "VOLUME" && unit.Volume.Type != "DELETION" {
		logrus.WithField("type", unit.Type).Debug("volume type 不需要执行,直接成功")
		return w.Client.Update(context.TODO(), &engine.Task{
			TaskId:     unit.TaskInstanceId,
			Status:     engine.Succeed,
			ExitCode:   0,
			ResultFile: "",
			ErrorMsg:   "",
		})
	}

	switch unit.Type {
	case "TASK":
		fallthrough
	case "VOLUME":
		if err := w.runTask(ctx, unit, unit.Runners[0]); err != nil {
			return err
		}
	case "RUN":
		log.Error("RUN 类型不支持,taskId: %s", unit.TaskInstanceId)
	case "DELETE":
		log.Error("DELETE 类型不支持,taskId: %s", unit.TaskInstanceId)
	default:
		log.Debugln("wrong format unit")
		return nil
	}
	return nil
}

func (w *Worker) runTask(ctx context.Context, unit *engine.Unit, runner *engine.Runner) error {
	// 添加资源限制
	w.setResources(unit)

	ctx, cancle := context.WithCancel(ctx)
	defer cancle()

	stopJob := make(chan struct{}, 2)

	go func() {
		var err error
		var stop bool
		for {
			stop, err = w.Client.Watch(ctx, unit.TaskInstanceId)
			if err != nil {
				logrus.WithError(err).Errorf("cannot watch task: %s\n", unit.TaskInstanceId)
			}
			if stop {
				logrus.Infof("任务被取消taskId: %s\n", unit.TaskInstanceId)
				cancle()
			}
			stopJob <- struct{}{}
			return
		}
	}()

	var err error
	finishCh := make(chan struct{}, 2)

	go func() {
		defer func() {
			err := recover()
			if err != nil {
				logrus.Errorf("runTask panic: %v\n", err)
				stackStr := string(debug.Stack())
				fmt.Println(stackStr)
			}
		}()

		err = w.runTaskInner(ctx, unit, runner)
		stopJob <- struct{}{}
		finishCh <- struct{}{}
	}()

	<-stopJob
	<-finishCh
	return err
}

func (w *Worker) runTaskInner(ctx context.Context, unit *engine.Unit, runner *engine.Runner) error {
	task := &engine.Task{
		TaskId:     runner.ID,
		ExitCode:   0,
		ResultFile: "",
		ErrorMsg:   "",
		WorkData:   unit.WorkerData,
	}

	unit.Log.Write([]byte(fmt.Sprintf("\n## ** "+time.Now().Format("2006-01-02 15:04:05")) + "开始创建pod\n\n"))

	unit.Log.Write([]byte(fmt.Sprintf("## 任务被执行器接收 workerId:%s workerName:%s workerType:%s workerId:%s\n",
		w.ID, w.Name, w.Type, w.ID)))

	unit.Log.Write([]byte(fmt.Sprintf("## 任务结果文件路径:%s\n", unit.ResultFile)))

	err := w.createPod(ctx, unit, runner)
	if err != nil {
		task.Status = engine.Failed
		task.ErrorMsg = "create pod err:" + err.Error()
		writeResult(unit.Log, task)
		w.Client.Update(context.TODO(), task)
		return err
	}

	return w.RunTaskReal(ctx, unit, runner)
}

func (w *Worker) RunTaskReal(ctx context.Context, unit *engine.Unit, runner *engine.Runner) error {
	task := &engine.Task{
		TaskId:     runner.ID,
		ExitCode:   0,
		ResultFile: "",
		ErrorMsg:   "",
		WorkData:   unit.WorkerData,
	}

	// 运行状态
	task.Status = engine.Running
	err := w.Client.Update(context.TODO(), task)
	if err != nil {
		logrus.WithError(err).
			WithField("taskId", task.TaskId).
			Error("cannot update task")
	}

	defer func() {
		w.Engine.DeletePod(context.TODO(), unit)
		w.Engine.DeleteCm(context.TODO(), unit)
	}()

	_, err, successed := w.Engine.WaitPodRunning(ctx, unit.PodSpec.Name, unit.PodSpec.Namespace, unit.Log)
	unit.Log.Write([]byte(fmt.Sprintf("\n## ** "+time.Now().Format("2006-01-02 15:04:05")) + "开始执行任务\n\n"))

	nodeName, _ := w.Engine.GetPodNodeName(unit.PodSpec.Name, unit.PodSpec.Namespace)
	task.WorkData = nodeName

	if err != nil && !successed && nodeName == "" {
		task.Status = engine.Failed
		task.ErrorMsg = "等待pod running err: " + err.Error()
		writeResult(unit.Log, task)
		w.Client.Update(context.TODO(), task)
		return err
	}

	if successed {
		task.Status = engine.Succeed
		writeResult(unit.Log, task)
		w.Client.Update(context.TODO(), task)
		return err
	}

	var e error
	readLogCh := make(chan struct{})
	go func() {
		logrus.Infof("开始读取日志,pod %s/%s\n", unit.PodSpec.Namespace, unit.PodSpec.Name)
		e = w.Engine.FetchLogs(ctx, unit, runner, unit.Log)
		logrus.Infof("读取日志完毕,pod %s/%s\n", unit.PodSpec.Namespace, unit.PodSpec.Name)
		if e != nil {
			logrus.WithError(e).Error("获取日志失败")
		}
		readLogCh <- struct{}{}
	}()

	code, err := w.Engine.WaitContainerNameTermination(ctx, "build", unit.PodSpec.Name, unit.PodSpec.Namespace)
	if err != nil || code != 0 {
		// 任务被取消了
		if err != nil && err == context.Canceled {
			task.Status = engine.Failed
			task.ExitCode = -1
			task.ResultFile = ""
			task.ErrorMsg = "任务被取消"
			writeResult(unit.Log, task)
			err := w.Client.Update(context.TODO(), task)
			if err != nil {
				logrus.WithError(err).
					WithField("taskId", task.TaskId).
					Error("取消任务更新失败")
			}
			return nil
		}
		if err != nil {
			task.ErrorMsg = "等待pod结束err: " + err.Error()
		} else {
			task.ErrorMsg = "pod结束,退出码异常 "
		}
		task.Status = engine.Failed
		task.ExitCode = int(code)
		task.ResultFile = ""
	} else {
		task.Status = engine.Succeed
		task.ExitCode = int(code)
		task.ResultFile = ""
	}

	// 等待日志读取
	<-readLogCh
	unit.Log.Write([]byte(fmt.Sprintf("\n## ** "+time.Now().Format("2006-01-02 15:04:05")) + "任务执行完毕\n\n"))
	// 获取resultFile
	if unit.ResultFile != "" && path.Base(unit.ResultFile) != "" {
		unit.Log.Write([]byte(fmt.Sprintf("## 开始读取结果文件:%s\n", unit.ResultFile)))
		select {
		case <-time.After(time.Second * 15):
		default:
			buffer := bytes.Buffer{}
			err = w.Engine.CopyFromContainer(context.TODO(), unit.ResultFile, "helper", unit.PodSpec.Name, unit.PodSpec.Namespace, &buffer)
			if err != nil {
				unit.Log.Write([]byte(fmt.Sprintf("## 读取结果文件:%s失败\n", unit.ResultFile)))
				task.ErrorMsg = fmt.Sprintf("读取结果文件:%s失败,err:%v\n", unit.ResultFile, err)
				task.Status = engine.Failed
			} else {
				unit.Log.Write([]byte(fmt.Sprintf("## 读取结果文件:%s成功\n", unit.ResultFile)))
				logrus.Debugf("read result file: %v,context:%s", unit.ResultFile, buffer.String())
				task.ResultFile = buffer.String()
				task.Status = engine.Succeed
			}
		}
	} else {
		unit.Log.Write([]byte(fmt.Sprintf("## 无结果文件\n")))
	}

	writeResult(unit.Log, task)

	var cmd string
	if unit.Volume.Type == "DELETION" {
		cmd = fmt.Sprintf("rm -rf %s;", path.Join(unit.Volume.Name, "*"))
	} else {
		if unit.ResultFile != "" {
			cmd = fmt.Sprintf("echo -n >%s;", unit.ResultFile)
		}
	}

	if cmd != "" {
		logrus.WithField("cmd", cmd).Debug("执行命令")
		buffer := bytes.Buffer{}
		_ = w.Engine.ExecCmd(context.TODO(), []string{
			"/bin/sh",
			"-c",
			cmd,
		}, "helper",
			unit.PodSpec.Name, unit.PodSpec.Namespace, &buffer)
		logrus.WithField("cmd", cmd).
			WithField("result", buffer.String()).
			Debug("执行命令完毕")
	}

	w.Client.Update(context.TODO(), task)
	return nil
}

func writeResult(w io.Writer, task *engine.Task) {
	w.Write([]byte(fmt.Sprintf("## 退出码:%d 执行状态:%s Msg:%s resultFile:%s\n", task.ExitCode, task.Status,
		task.ErrorMsg, task.ResultFile)))
}

func (w *Worker) createPod(ctx context.Context, unit *engine.Unit, runner *engine.Runner) error {
	log := logger.FromContext(ctx).
		WithField("pod.name", unit.PodSpec.Name)

	if err := w.Engine.CreatePod(ctx, unit); err != nil {
		log.Debugln("创建Pod失败")
		return err
	}

	return nil
}

func (w *Worker) setResources(unit *engine.Unit) {
	for _, runner := range unit.Runners {
		if runner.Name == "helper" {
			runner.Resources = w.Resources.HelperResources
		} else if runner.Name == "build" {
			runner.Resources = w.Resources.BuildResources
		}
	}
}

// 从Runner返回密钥数组
func secretSlice(runner *engine.Runner) []types.Secret {
	var secrets []types.Secret
	for i := 0; i < runner.GetSecretLen(); i++ {
		secrets = append(secrets, runner.GetSecretAt(i))
	}
	return secrets
}
